package org.jheaps.dag

import org.jheaps.AddressableHeap
import org.jheaps.MergeableAddressableHeap
import org.jheaps.annotations.ConstantTime
import org.jheaps.annotations.LogarithmicTime
import java.io.Serializable
import kotlin.math.max

/**
 * Hollow heaps. The heap is sorted according to the [ natural ordering][Comparable] of its keys, or by a [Comparator] provided at heap
 * creation time, depending on which constructor is used.
 *
 *
 *
 * This is the hollow heap described in detail in the following
 * [paper](https://doi.org/10.1145/3093240):
 *
 *  * TD Hansen, H Kaplan, RE Tarjan and U Zwick. Hollow heaps. ACM
 * Transactions on Algorithms (TALG), 13(3), 42, 2017.
 *
 *
 *
 *
 * This implementation provides amortized O(1) time for operations that do not
 * involve deleting an element such as `insert`, and `decreaseKey`.
 * Operations `deleteMin` and `delete` are amortized O(log(n)). The
 * operation `meld` is also amortized O(1).
 *
 *
 *
 * All the above bounds, however, assume that the user does not perform
 * cascading melds on heaps such as:
 *
 * <pre>
 * d.meld(e);
 * c.meld(d);
 * b.meld(c);
 * a.meld(b);
</pre> *
 *
 * The above scenario, although efficiently supported by using union-find with
 * path compression, invalidates the claimed bounds.
 *
 *
 *
 * Note that the ordering maintained by this heap, like any heap, and whether or
 * not an explicit comparator is provided, must be *consistent with
 * `equals`* if this heap is to correctly implement the
 * `AddressableHeap` interface. (See `Comparable` or
 * `Comparator` for a precise definition of *consistent with
 * equals*.) This is so because the `AddressableHeap` interface is
 * defined in terms of the `equals` operation, but this heap performs all
 * key comparisons using its `compareTo` (or `compare`) method, so
 * two keys that are deemed equal by this method are, from the standpoint of
 * this heap, equal. The behavior of a heap *is* well-defined even if its
 * ordering is inconsistent with `equals`; it just fails to obey the
 * general contract of the `AddressableHeap` interface.
 *
 *
 *
 * **Note that this implementation is not synchronized.** If
 * multiple threads access a heap concurrently, and at least one of the threads
 * modifies the heap structurally, it *must* be synchronized externally.
 * (A structural modification is any operation that adds or deletes one or more
 * elements or changing the key of some element.) This is typically accomplished
 * by synchronizing on some object that naturally encapsulates the heap.
 *
 * @param <K>
 * the type of keys maintained by this heap
 * @param <V>
 * the type of values maintained by this heap
 *
 * @author Dimitrios Michail
</V></K> */
class HollowHeap<K, V> @ConstantTime constructor(
    /**
     * The comparator used to maintain order in this heap, or null if it uses
     * the natural ordering of its keys.
     *
     * @serial
     */
    private val comparator: Comparator<in K?>?
) : MergeableAddressableHeap<K, V>,
    Serializable {

    /**
     * The last node in the root list
     */
    private var root: HollowNode<K, V>? = null

    /**
     * Size of the heap
     */
    private var size: Long

    /**
     * Number of nodes (hollow or not). Useful for rebuilding.
     */
    private var nodes: Long

    /**
     * Auxiliary array for performing links.
     */
    private val aux: Array<HollowNode<K, V>?>

    /**
     * Used to reference the current heap or some other pairing heap in case of
     * melding, so that handles remain valid even after a meld, without having
     * to iterate over them.
     *
     * In order to avoid maintaining a full-fledged union-find data structure,
     * we disallow a heap to be used in melding more than once. We use however,
     * path-compression in case of cascading melds, that it, a handle moves from
     * one heap to another and then another.
     */
    private var other: HollowHeap<K, V>

    /**
     * Constructs a new, empty heap, using the natural ordering of its keys. All
     * keys inserted into the heap must implement the [Comparable]
     * interface. Furthermore, all such keys must be *mutually
     * comparable*: `k1.compareTo(k2)` must not throw a
     * `ClassCastException` for any keys `k1` and `k2` in the
     * heap. If the user attempts to put a key into the heap that violates this
     * constraint (for example, the user attempts to put a string key into a
     * heap whose keys are integers), the `insert(Object key)` call will
     * throw a `ClassCastException`.
     */
    @ConstantTime
    constructor() : this(null)

    /**
     * Constructs a new, empty heap, ordered according to the given comparator.
     * All keys inserted into the heap must be *mutually comparable* by
     * the given comparator: `comparator.compare(k1,
     * k2)` must not throw a `ClassCastException` for any keys `k1`
     * and `k2` in the heap. If the user attempts to put a key into the
     * heap that violates this constraint, the `insert(Object key)` call
     * will throw a `ClassCastException`.
     *
     * @param comparator
     * the comparator that will be used to order this heap. If
     * `null`, the [natural ordering][Comparable] of
     * the keys will be used.
     */
    init {
        this.size = 0
        this.nodes = 0
        this.aux = arrayOfNulls(AUX_BUCKET_ARRAY_SIZE)
            /*java.lang.reflect.Array.newInstance(
            HollowNode::class.java,
            AUX_BUCKET_ARRAY_SIZE
        ) as Array<HollowNode<K, V>?>*/
        this.other = this
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @ConstantTime(amortized = true)
    override fun insert(key: K, value: V?): AddressableHeap.Handle<K, V> {
        check(other === this) { "A heap cannot be used after a meld" }
        if (key == null) {
            throw NullPointerException("Null keys not permitted")
        }

        val e = Item<K, V>(key, value)
        val u = HollowNode(this, key)
        u.item = e
        e.node = u
        nodes++

        root = if (root == null) {
            u
        } else {
            link(root!!, u)
        }

        size++
        return e
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @ConstantTime(amortized = true)
    override fun insert(key: K): AddressableHeap.Handle<K, V> {
        return insert(key, null)
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime(amortized = false)
    override fun findMin(): Item<K, V> {
        if (size == 0L) {
            throw NoSuchElementException()
        }
        return root!!.item!!
    }

    /**
     * {@inheritDoc}
     */
    @LogarithmicTime(amortized = true)
    override fun deleteMin(): AddressableHeap.Handle<K, V> {
        if (size == 0L) {
            throw NoSuchElementException()
        }
        val item = root!!.item
        item!!.delete()
        return item
    }

    @get:ConstantTime
    override val isEmpty: Boolean
        /**
         * {@inheritDoc}
         */
        get() = size == 0L

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun size(): Long {
        return size
    }

    /**
     * {@inheritDoc}
     */
    override fun comparator(): Comparator<in K?>? {
        return comparator
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime(amortized = false)
    override fun clear() {
        root = null
        size = 0
        nodes = 0
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun meld(other: MergeableAddressableHeap<K, V>) {
        val h = other as HollowHeap<K, V>

        // check same comparator
        if (comparator != null) {
            require(!(h.comparator == null || h.comparator != comparator)) { "Cannot meld heaps using different comparators!" }
        } else require(h.comparator == null) { "Cannot meld heaps using different comparators!" }

        check(h.other === h) { "A heap cannot be used after a meld." }

        // meld
        if (root == null) {
            root = h.root
        } else if (h.root != null) {
            root = link(root!!, h.root!!)
        }
        size += h.size
        nodes += h.nodes

        // clear other
        h.size = 0
        h.nodes = 0
        h.root = null

        // take ownership
        h.other = this
    }

    // --------------------------------------------------------------------
    class Item<K, V>(override var key: K, override var value: V?) : AddressableHeap.Handle<K, V>, Serializable {
        var node: HollowNode<K, V>? = null

        override fun decreaseKey(newKey: K) {
            checkInvalid()
            owner.decreaseKey(this, newKey)
        }

        override fun delete() {
            checkInvalid()
            owner.delete(this)
        }

        val owner: HollowHeap<K, V>
            get() = node!!.owner

        private fun checkInvalid() {
            requireNotNull(node) { "Invalid handle!" }
        }

    }

    class HollowNode<K, V>(/*
         * We maintain explicitly the belonging heap, instead of using an inner
         * class due to possible cascading melding.
         */var heap: HollowHeap<K, V>, var key: K
    ) : Serializable {
        var child: HollowNode<K, V>? = null // child
        var next: HollowNode<K, V>? = null // next sibling in list of children of first

        // parent
        var sp: HollowNode<K, V>? = null // second parent
        var rank: Int = 0

        /*
         * The item inside the node. If null the node is hollow.
         */
        var item: Item<K, V>? = null

        val owner: HollowHeap<K, V>
            /*
         * Get the owner heap of the handle. This is union-find with
         * path-compression between heaps.
         */
            get() {
                if (heap.other !== heap) {
                    // find root
                    var root = heap
                    while (root !== root.other) {
                        root = root.other
                    }
                    // path-compression
                    var cur = heap
                    while (cur.other !== root) {
                        val next = cur.other
                        cur.other = root
                        cur = next
                    }
                    heap = root
                }
                return heap
            }
    }

    /**
     * Decrease the key of an item.
     *
     * @param n
     * the item
     * @param newKey
     * the new key
     */
    private fun decreaseKey(e: Item<K, V>, newKey: K) {
        assert(e.node != null)
        val c = comparator?.compare(newKey, e.key) ?: (newKey as Comparable<K>).compareTo(e.key)
        require(c <= 0) { "Keys can only be decreased!" }

        val u = e.node
        if (c == 0 || u === root) {
            e.key = newKey
            u!!.key = newKey
            return
        }

        // move item to new node
        val v = HollowNode(this, newKey)
        nodes++
        u!!.item = null
        v.item = e
        e.node = v
        e.key = newKey

        if (u.rank > 2) {
            v.rank = u.rank - 2
        }
        v.child = u
        u.sp = v

        // link new node with root
        root = link(root!!, v)
    }

    /**
     * Delete an item
     *
     * @param e
     * the item
     */
    private fun delete(e: Item<K, V>) {
        assert(e.node != null)
        assert(e.node!!.item === e)
        // delete item
        e.node!!.item = null
        e.node = null
        size--

        if (root!!.item != null) {
            // non-minimum deletion
            return
        }

        // minimum deletion
        var maxRank = -1
        while (root != null) {
            val v  = root!!
            root = root!!.next

            var w = v.child
            while (w != null) {
                val u = w
                w = w.next
                u.next = null

                if (u.item == null) { // hollow
                    if (u.sp == null) { // v is the only parent
                        u.next = root
                        root = u
                    } else { // two parents
                        if (u.sp === v) { // v is the second parent
                            u.sp = null
                        } else { // v is the first parent
                            u.sp = null
                            u.next = null
                        }
                    }
                } else {
                    maxRank = max(maxRank.toDouble(), doRankedLinks(u).toDouble()).toInt()
                }
            }
            nodes-- // garbage collect v
            v.next = null
            v.child = null
            v.sp = null
            v.item = null
        }
        doUnrankedLinks(maxRank)
    }

    private fun doRankedLinks(u: HollowNode<K, V>): Int {
        var u = u
        while (aux[u.rank] != null) {
            u = link(u, aux[u.rank]!!)
            aux[u.rank] = null
            u.rank += 1
        }
        aux[u.rank] = u
        return u.rank
    }

    private fun doUnrankedLinks(maxRank: Int) {
        assert(root == null)
        for (i in 0..maxRank) {
            val u = aux[i]
            if (u != null) {
                root = if (root == null) {
                    u
                } else {
                    link(root!!, u)
                }
                aux[i] = null
            }
        }
    }

    private fun link(v: HollowNode<K, V>, w: HollowNode<K, V>): HollowNode<K, V> {
        val c = comparator?.compare(v.key, w.key) ?: (v.key as Comparable<K?>).compareTo(w.key)
        if (c > 0) {
            // v a child of w
            v.next = w.child
            w.child = v
            return w
        } else {
            // w a child of v
            w.next = v.child
            v.child = w
            return v
        }
    }

    companion object {
        /**
         * Size of bucket array. Based on maximum rank.
         */
        private const val AUX_BUCKET_ARRAY_SIZE = 128
    }
}
